// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#pragma once

#include <deque>
#include <functional>
#include <optional>
#include <utility>

#include <boost/call_traits.hpp>
#include <glog/logging.h>

namespace math {
namespace detail_tmw {

struct NullHandler {
  template <typename P1, typename P2>
  void OnPushed(P1&&, P2&&) {}
  template <typename P1, typename P2>
  void OnPopped(P1&&, P2&&) {}
  void OnClear() {}
};

template <typename Handler>
struct PassHandler {
  PassHandler() : handler(nullptr) {}
  explicit PassHandler(Handler* handler) : handler(handler) {}

  template <typename P1, typename P2>
  void OnPushed(P1&& time, P2&& value) {
    handler->OnPushed(std::forward<P1>(time), std::forward<P2>(value));
  }

  template <typename P1, typename P2>
  void OnPopped(P1&& time, P2&& value) {
    handler->OnPopped(std::forward<P1>(time), std::forward<P2>(value));
  }

  void OnClear() { handler->OnClear(); }

  Handler* handler;
};

template <typename T, typename Handler, typename Time = int64_t>
struct LastValueWrapper : public Handler {
  using ValueType = T;
  using TimeType = Time;
  using ValueParamType = typename boost::call_traits<ValueType>::param_type;
  using TimeParamType = typename boost::call_traits<TimeType>::param_type;
  using TimeValuePairType = std::pair<TimeType, ValueType>;

  template <typename... Args>
  explicit LastValueWrapper(Args&&... args) : Handler(std::forward<Args>(args)...) {}

  void OnPushed(TimeParamType time, ValueParamType value) {
    Handler::OnPushed(time, value);
    last_pushed = TimeValuePairType(time, value);
  }

  void OnPopped(TimeParamType time, ValueParamType value) {
    last_popped = TimeValuePairType(time, value);
    Handler::OnPopped(time, value);
  }

  template <typename P>
  void OnPopped(TimeParamType time, P&& value) {
    last_popped = TimeValuePairType(time, std::move(value));
    Handler::OnPopped(time, last_popped.value().second);
  }

  void OnClear() {
    last_pushed = std::nullopt;
    last_popped = std::nullopt;
    Handler::OnClear();
  }

  std::optional<TimeValuePairType> last_pushed;
  std::optional<TimeValuePairType> last_popped;
};

template <typename T, typename Handler, typename Time = int64_t>
struct FromMovingWindowHandler : public Handler {
  using ValueParamType = typename boost::call_traits<T>::param_type;
  using TimeParamType = typename boost::call_traits<Time>::param_type;

  template <typename... Args>
  explicit FromMovingWindowHandler(Args&&... args) : Handler(std::forward<Args>(args)...) {}

  void OnPushed(TimeParamType, ValueParamType value) { Handler::OnPushed(value); }

  void OnPopped(TimeParamType, ValueParamType value) { Handler::OnPopped(value); }

  template <typename P>
  void OnPopped(TimeParamType, P&& value) {
    Handler::OnPopped(std::move(value));
  }

  void OnClear() { Handler::OnClear(); }
};

}  // namespace detail_tmw

template <typename T, typename Handler, typename Time = int64_t, typename Duration = int64_t>
class TimeMovingWindow {
 public:
  using ValueType = T;
  using TimeType = Time;
  using DurationType = Duration;

  using ValueParamType = typename boost::call_traits<ValueType>::param_type;
  using TimeParamType = typename boost::call_traits<TimeType>::param_type;
  using DurationParamType = typename boost::call_traits<DurationType>::param_type;

  using TimeValuePairType = std::pair<TimeType, ValueType>;
  using Container = std::deque<TimeValuePairType>;
  using Iterator = typename Container::iterator;
  using ConstIterator = typename Container::const_iterator;

  template <typename... Args>
  explicit TimeMovingWindow(Args&&... args)
      : handler_(std::forward<Args>(args)...), window_size_(DurationType()) {}

  ~TimeMovingWindow() {}

  void Push(TimeParamType time, ValueParamType value) {
    window_.emplace_back(time, value);
    handler_.OnPushed(time, value);
    PushTime(time);
  }

  template <typename... Args>
  void Emplace(TimeParamType time, Args&&... args) {
    window_.emplace_back(time, ValueType(std::forward<Args>(args)...));
    handler_.OnPushed(time, window_.back().second);
    PushTime(time);
  }

  void PushTime(TimeParamType time) {
    bool empty = true;
    for (auto it = window_.begin(); it != window_.end(); ++it) {
      if (time - it->first < window_size_) {
        empty = false;
        if (it != window_.begin()) {
          window_.erase(window_.begin(), it);
        }
        break;
      } else {
        ++num_popped_;
        handler_.OnPopped(it->first, std::move(it->second));
      }
    }
    if (empty) window_.clear();
  }

  void ResizeWindow(DurationParamType window_size) {
    window_size_ = window_size;
    if (!empty()) {
      PushTime(window_.back().first);
    }
  }

  void Clear() {
    window_.clear();
    handler_.OnClear();
  }

  size_t num_popped() const { return num_popped_; }
  DurationParamType window_size() const { return window_size_; }

  bool empty() const { return window_.empty(); }
  size_t size() const { return window_.size(); }

  const TimeValuePairType& front() const { return window_.front(); }
  const TimeValuePairType& back() const { return window_.back(); }

  Iterator begin() { return window_.begin(); }
  Iterator end() { return window_.end(); }
  ConstIterator begin() const { return window_.begin(); }
  ConstIterator end() const { return window_.end(); }

  Handler& handler() { return handler_; }
  const Handler& handler() const { return handler_; }

 private:
  Handler handler_;
  DurationType window_size_;
  size_t num_popped_ = 0;
  Container window_;
};

template <
    typename T,
    bool LastValueStore = false,
    typename Time = int64_t,
    typename Duration = int64_t>
class SimpleTimeMovingWindow;

template <typename T, typename Time, typename Duration>
class SimpleTimeMovingWindow<T, false, Time, Duration>
    : public TimeMovingWindow<T, detail_tmw::NullHandler, Time, Duration> {};

template <typename T, typename Time, typename Duration>
class SimpleTimeMovingWindow<T, true, Time, Duration>
    : public TimeMovingWindow<
          T,
          detail_tmw::LastValueWrapper<T, detail_tmw::NullHandler, Time>,
          Time,
          Duration> {};

template <typename T, typename Handler, typename Time = int64_t, typename Duration = int64_t>
class LastValueStoreTimeMovingWindow
    : public TimeMovingWindow<T, detail_tmw::LastValueWrapper<T, Handler, Time>, Time, Duration> {};

template <typename T, typename Compare, typename Time = int64_t, typename Duration = int64_t>
class TimeMovingWindowMinMax {
 public:
  using ValueType = T;
  using TimeType = Time;
  using DurationType = Duration;

  using ValueParamType = typename boost::call_traits<ValueType>::param_type;
  using TimeParamType = typename boost::call_traits<TimeType>::param_type;
  using DurationParamType = typename boost::call_traits<DurationType>::param_type;

  using TimeValuePairType = std::pair<TimeType, ValueType>;
  using Container = std::deque<TimeValuePairType>;

  void ResizeWindow(DurationParamType window_size) {
    window_size_ = window_size;
    if (!window_.empty()) {
      PushTime(window_.back().first);
    }
  }

  void Push(TimeParamType time, ValueParamType value) {
    PushTime(time);

    Compare compare;
    bool clear = true;
    for (auto r_it = window_.rbegin(); r_it != window_.rend(); ++r_it) {
      if (compare(r_it->second, value)) {
        window_.erase(r_it.base(), window_.end());
        window_.emplace_back(time, value);
        clear = false;
        break;
      }
    }

    if (clear) {
      window_.clear();
      window_.emplace_back(time, value);
    }
  }

  void PushTime(TimeParamType time) {
    bool empty = true;
    for (auto it = window_.begin(); it != window_.end(); ++it) {
      if (time - it->first < window_size_) {
        empty = false;
        if (it != window_.begin()) {
          window_.erase(window_.begin(), it);
          popped_ = true;
        }
        break;
      }
    }
    if (empty && !window_.empty()) {
      window_.clear();
      popped_ = true;
    }
  }

  DurationParamType window_size() const { return window_size_; }
  bool popped() const { return popped_; }

  bool has_value() const { return !window_.empty(); }
  const ValueParamType value() const { return window_.front().second; }

 private:
  DurationType window_size_;
  bool popped_ = false;
  Container window_;
};

template <typename T, typename Time = int64_t, typename Duration = int64_t>
using TimeMovingWindowMin = TimeMovingWindowMinMax<T, std::less<T>, Time, Duration>;

template <typename T, typename Time = int64_t, typename Duration = int64_t>
using TimeMovingWindowMax = TimeMovingWindowMinMax<T, std::greater<T>, Time, Duration>;

}  // namespace math
